package com.bungarus.busi;

import com.bungarus.Access;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.*;
import org.apache.log4j.Logger;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by tang on 2018/11/18.
 */
public class AccessRemedy extends Remedy {
    Logger logger = Logger.getLogger(AccessRemedy.class);
    private Map<String, List<InnerClient>> backendClients;
    public static void start(BungarusServer server0) throws Exception{
        if(!started) {
            started = true;
            new AccessRemedy().run(server0);
        }
    }

    @Override
    protected void init(BungarusServer server){
        assert server instanceof Access;
        backendClients = ((Access)server).getInnerClientList();
    }

    @Override
    protected void process(){
        DischargeQueue queue = Discharger.getQueueInstance(); //get generic queue
        AccessRemedy.this.retry0(queue);

        queue = Discharger.getQueueInstance(Discharger.DISCHARGE_NAME_CLIENT); //get client queue
        AccessRemedy.this.retry(queue);
    }

    @Override
    protected void retry0(DischargeQueue queue) {
        logger.debug("Size of generic discharge queue: " + queue.size());
        if(queue.size() <= 0 || backendClients.isEmpty()){
            return;
        }
        ConcurrentHashMap.KeySetView<Long, MessageProtobuf.Message> keys = queue.keys();
        for(Long l: keys){
            MessageProtobuf.Message noAckMsg = queue.getFromQueue(l);
            if((System.currentTimeMillis() - noAckMsg.getHeader().getSentTime()) >= ACK_TIMEOUT) {
                List<InnerClient> innerClients = null;
                switch (noAckMsg.getHeader().getType()) {
                    case MessageType.IM:
                    case MessageType.IM_GROUP:
                        innerClients = backendClients.get(Access.CLUSTER_TAG_BUSI);
                        break;
                    case MessageType.STATUS:
                    case MessageType.AUTH:
                        innerClients = backendClients.get(Access.CLUSTER_TAG_TRANSFER);
                        break;
                    default:
                        break;
                }
                if(null != innerClients && innerClients.size() > 0) {
                    MessageProtobuf.Extension extension = noAckMsg.getHeader().getExtensions(0);
                    if(null != extension && extension.getKey().equalsIgnoreCase(MessageBuilder.TAG)) {
                        String tag = extension.getValue();
                        try {
                            // if no ack when timeout, remove the message first, and put it into the queue again
                            if(MessageStatus.REMEDY == noAckMsg.getHeader().getReturnCode()) {
                                innerClients.get(0).saveTryFailedMessage(noAckMsg);
                            }
                            else {
                                MessageProtobuf.Message msg = MessageBuilder.buildRemedyMessage(noAckMsg);
                                long index = Long.parseLong(msg.getHeader().getId()) % innerClients.size();
                                innerClients.get((int)index).sendMessage(msg, tag);
                            }
                            queue.removeFromQueue(l);
                        }
                        catch (BungarusClientException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }
    }

    /**
     * deal with the message for client
     * @param queue
     */
    private void retry(DischargeQueue queue) {
        logger.debug("Size of end user's client discharge queue: " + queue.size());
        ConcurrentHashMap.KeySetView<Long, MessageProtobuf.Message> keys = queue.keys();
        for(Long l: keys){
            MessageProtobuf.Message noAckMsg = queue.getFromQueue(l);
            if((System.currentTimeMillis() - noAckMsg.getHeader().getSentTime()) >= ACK_TIMEOUT) {
                try {
                    // if no ack when timeout, guess the target client is offline, then remove the session
                    List<InnerClient> innerClients = backendClients.get(Access.CLUSTER_TAG_BUSI);
                    long index = Long.parseLong(noAckMsg.getHeader().getId()) % innerClients.size();
                    innerClients.get((int)index).sendMessage(
                            MessageBuilder.buildOfflineMessage(noAckMsg),
                            Access.CLUSTER_TAG_BUSI);
                    queue.removeFromQueue(l);
                    //clear session because of offline message
                    Session session = new Session();
                    session.setId(noAckMsg.getHeader().getSessionId());
                    Access.getSessionContainer().remove(session);
                }
                catch (BungarusClientException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
